package com.simple.datax.conn.impl;

import java.io.UnsupportedEncodingException;

import javax.jms.MessageListener;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.SimpleMessage;
import com.simple.datax.api.ConnectorException;

import ionic.Msmq.Message;
import ionic.Msmq.MessageQueueException;
import ionic.Msmq.TransactionType;
import ionic.Msmq.Queue;

/** 
 * @Title: MSMQConnImpl.java 
 * @Package com.simple.datax.conn.impl 
 * @Description: 实现具体的MS MQ连接操作，包括连接，接收，发送等。 
 * @author Jin Yi 
 * @date 2012-12-30
 * @version V1.0 
 */
public class MSMQConnImpl extends MQBaseConnector {
	private static final Logger logger = LoggerFactory
			.getLogger(MSMQConnImpl.class);
	public static final String TYPE = "MSMQ";
	private Queue queue= null;
	private TransactionType type = null;
	private final int timeout = 2000;
	private String conn;
	
	private long recvTotal;
	private long sendTotal;
	
	/**
	 * 构造器
	 * @param connPara
	 * @throws ConnectorException
	 */
	public MSMQConnImpl(String connPara) throws ConnectorException {
		super(connPara);
		this.connPara = connPara;
		parseConnPara();
	}
	
	private void parseConnPara() throws ConnectorException {
		String[] tmp = splitConnPara(connPara);
		if (tmp.length == 2) {
//			char dollar = connPara.charAt(1);
//		if (dollar == '$') {
//			String tx = connPara.substring(0,1);
			int i = Integer.valueOf(tmp[0]);
			switch(i) {
			case 1:
				type = TransactionType.None;
				break;
			case 2:
				type = TransactionType.SINGLE_MESSAGE;
				break;
			case 3:
				type = TransactionType.XA;
				break;
			case 4:
				type = TransactionType.MTS;
				break;
			}
			conn = tmp[1];
		} else {
			throw new ConnectorException("Connector failed");
		}
	}
	
	@Override
	public String getMessage() throws ConnectorException {
		String ret = null;
		prepareQueue();
		try {
            Message msg= queue.receive(timeout);
//            logger.info(" No Convert: " + new String(msg.getBody()));
            ret = new String(msg.getBody(), getEncoding());
//            logger.info(" Convert: " + ret);
            logger.debug("########Receiver Get Message: #" + ++recvTotal + " - " + connPara);
        }
        catch (MessageQueueException ex) {
        	if (ex.hresult == 0xC00E001B) {
        		logger.warn("Receive timeout");
        	} else {
        		logger.warn("Receive failed", ex);
        		
        		// Re-try after create new queue connection
        		queue = null;
        		prepareQueue();
        		try {
	        		Message msg= queue.receive(timeout);
	        		
	        		ret = new String(msg.getBody(), getEncoding());
	                logger.debug("########Receiver Get Message: #" + ++recvTotal + " - " + connPara);
        		} catch (MessageQueueException e) {
        			logger.error("Receive re-try failded", e);
        		} catch (UnsupportedEncodingException e) {
        			logger.error("########Receiver Get Message Failed: #" + ++recvTotal + " - " + connPara);
                	throw new ConnectorException("Get message from MSMQ failed", ex);
				}
        	}
        } catch (UnsupportedEncodingException ex) {
        	logger.info("########Receiver Get Message Failed: #" + ++recvTotal + " - " + connPara);
        	throw new ConnectorException("Get message from MSMQ failed", ex);
		}
		return ret;
	}
	
	/**
	 * 检查队列并为操作做准备
	 * @throws ConnectorException
	 */
	private void prepareQueue() throws ConnectorException {
		try {
			if (queue == null) {
				queue = new Queue(conn);
				logger.debug("open: OK.");
			}
		} catch (MessageQueueException ex1) {
			throw new ConnectorException("Open queue failed", ex1);
		}
	}
    
	/**
	 * MSMQ 发送消息
	 * @param message
	 * @throws MessageQueueException
	 * @throws UnsupportedEncodingException
	 */
    private void send(String message) throws MessageQueueException, UnsupportedEncodingException {
    	try {
            // the transaction flag must agree with the transactional flavor of the queue.
//    		String content = getOutputString(message, getEncoding());
//    		logger.debug("Step" + message);
//    		logger.debug("Step" + content);
            Message msg= new Message(message.getBytes(getEncoding()), "label", "L:none");
//            msg.setBody(message.getBytes(getEncoding()));
            queue.send(msg, type);
            logger.debug("Sent out message");
        }
        catch (MessageQueueException ex) {
        	throw ex;
        } catch (UnsupportedEncodingException ex) {
        	throw ex;
		}
    }
    
    @Override
	protected void prepareSend() throws ConnectorException {
    	prepareQueue();
	}

	@Override
	protected void internalSend(SimpleMessage message) throws ConnectorException {
		try {
			/**
			 * MSMQ has outgoing queue to cache messages when network is broken.
			 * It could re-send when network is connected. so from code side, can
			 * not catch the exception for network issue, and suppose that message
			 * will reach target queue at the end.
			 * Please refer to resource to know more detail
			 * http://blogs.msdn.com/b/johnbreakwell/archive/2008/11/12/how-do-i-create-an-msmq-outgoing-queue.aspx
			 */
			send(message.getContent());
			logger.debug("########Sender Sent Message: #" + ++sendTotal + " - " + connPara);
		} catch (MessageQueueException ex) {
			logger.error("########Sender Sent Message Failed: #" + ++sendTotal + " - " + connPara, ex);
			throw new ConnectorException("MSMQ internalSend failed", ex, true);
		} catch (Exception ex) {
			logger.error("########Sender Sent Message Failed: #" + ++sendTotal + " - " + connPara, ex);
			throw new ConnectorException("MSMQ internalSend failed", ex);
		}
	}

	@Override
	public void setMessageListener(MessageListener listener)
			throws ConnectorException {
		// TODO Auto-generated method stub
		
	}
	
	@Override
	public String getConnType() {
		return TYPE;
	}

	@Override
	public void disconnect() {
		if (queue != null) {
			try {
				queue.close();
			} catch (MessageQueueException e) {
				logger.warn("MSMQ disconnect error", e);
			}
			queue = null;
		}
	}

	@Override
	protected void prepareReceive() throws ConnectorException {
		prepareQueue();
	}
}
